/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb.client;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.List;
import java.util.Properties;
import java.util.Random;
import java.util.UUID;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer;
import org.apache.log4j.Logger;
import org.json.JSONObject;

public class KfkProducer { 
	private String zkConnect;
	private int ackMode;
	private String id;
	private kafka.javaapi.producer.Producer<String, byte[]> producer;
	
    private static Logger LOG = Logger.getLogger(KfkProducer.class.getName());

    private Random rnd = new Random();
    
    public String getZkConnect() {
    	return zkConnect;
    }
    
    public int getAckMode() {
    	return ackMode;
    }
    
    public String getId() {
    	return id;
    }
    
    public KfkProducer(String zkConnect, int ackMode, String id) {
		this.zkConnect = zkConnect;
		this.ackMode = ackMode;
		if (id == null)
			this.id = UUID.randomUUID().toString().replace("-", "");
		else
			this.id = id;

		Properties props = createProducerProps();
		props.put("request.required.acks", String.valueOf(this.ackMode));
		ProducerConfig config = new ProducerConfig(props);
		producer = new kafka.javaapi.producer.
				Producer<String, byte[]>(config);    			
        LOG.info(this.id + ": producer created.");
	}
	
	public void send(String topic, String message) {
		if (topic != null && message != null) {
			String key = String.valueOf(rnd.nextInt());
			KeyedMessage<String, byte[]> kmsg = 
            	new KeyedMessage<String, byte[]>(topic, key, message.getBytes());
            producer.send(kmsg);
		} else {
			LOG.warn(id + ": sending to null topic or with message...");
		}
	}
	
    private Properties createProducerProps() {
        Properties props = new Properties();
        FileInputStream fis = null;
		try {
			fis = new FileInputStream("conf/csbproducer.properties");
			props.load(fis);
		} catch (IOException e) {			
		} finally {
			if (fis != null)
				try {
					fis.close();
				} catch (IOException e) {}
		}
		
    	int sessionTimeout = Integer.parseInt(props.getProperty("zookeeper.session.timeout.ms", "6000"));
    	props.remove("zookeeper.session.timeout.ms");
    	int connTimeout = Integer.parseInt(props.getProperty("zookeeper.connection.timeout.ms", "6000"));
    	props.remove("zookeeper.connection.timeout.ms");
    	
    	String blist = "";
    	ZkClient zkClient = new ZkClient(zkConnect, sessionTimeout, 
    			connTimeout, new BytesPushThroughSerializer());
    	List<String> brokers = zkClient.getChildren("/brokers/ids");
    	if (brokers != null && !brokers.isEmpty()) {
    		int n = brokers.size();
    		if (n > 3) n = 3;
    		for (int i = 0; i < n; i++) {
    			JSONObject jsonObj = new JSONObject
    				(new String((byte[])zkClient.readData
    					("/brokers/ids/" + brokers.get(i))));
    			String host = jsonObj.getString("host");
    			String port = String.valueOf(jsonObj.getInt("port"));
    			if (!host.isEmpty() && !port.isEmpty()) {
    				String conn = host + ":" + port;
    				if (blist.isEmpty()) blist = conn;
    				else blist = blist + "," + conn;
    			}
    		}
    	}
    	
        props.put("metadata.broker.list", blist);
        props.put("request.required.acks", "-1");
        props.put("producer.type", "sync");
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
        props.put("serializer.class", "kafka.serializer.DefaultEncoder");
        props.put("partitioner.class", "kafka.producer.DefaultPartitioner");
        props.put("client.id", KfkProducer.class.getName());
 
        return props;
    }
}

